package avicit.bdp.dcs.json.service.reader;

import avicit.bdp.common.utils.database.DBUtils;
import avicit.bdp.dcs.datasource.dto.JobDatasource;
import avicit.bdp.dcs.datasource.service.DataSourceUtils;
import avicit.bdp.dcs.job.dto.JobDTO;
import avicit.bdp.dcs.json.service.util.BuildJsonUtil;
import avicit.bdp.dcs.plugin.BaseDataxParam;
import avicit.bdp.dcs.plugin.odps.param.OdpsDataxParam;
import avicit.bdp.dcs.task.dto.TaskDTO;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;

@Component
public class OdpsReaderBuilder {

  @Autowired private DataSourceUtils dataSourceUtils;
  @Autowired private BuildJsonUtil buildJsonUtil;

  private static final String SRC_COLUMN_NAME = "srcColumnName";

  /** 构建ODPS采集Reader */
  public BaseDataxParam buildOdpsReaderByStructure(TaskDTO taskDTO, JobDTO jobDTO)
          throws Exception {
    JobDatasource readerDatasource = dataSourceUtils.convertDatasource(jobDTO.getSrcDatasourceId());

    OdpsDataxParam dataxParam = new OdpsDataxParam();
    dataxParam.setAccessId(readerDatasource.getJdbcUsername());
    dataxParam.setAccessKey(readerDatasource.getJdbcPassword());
    dataxParam.setProject(readerDatasource.getDatabaseName());
    dataxParam.setTable(DBUtils.convertTableName(readerDatasource.getDbType(), taskDTO.getSrcTableName()));
    dataxParam.setPartition(new ArrayList<>());
    dataxParam.setColumnList(getReaderColumns(taskDTO, readerDatasource.getDbType()));
    dataxParam.setPackageAuthorizedProject("");
    dataxParam.setSplitMode("record");
    dataxParam.setOdpsServer(readerDatasource.getBaseDataSource().getIp());

    return dataxParam;
  }

  private List<String> getReaderColumns(TaskDTO taskDTO, String dbType) {
    List<String> readerColumns = new ArrayList<>();
    JSONArray jsonArray = JSON.parseArray(taskDTO.getMappingColumn());
    for (int i = 0; i < jsonArray.size(); i++) {
      JSONObject jsonObject = jsonArray.getJSONObject(i);
      readerColumns.add(buildJsonUtil.convertColumnName(dbType, jsonObject.getString(SRC_COLUMN_NAME)));
    }

    return readerColumns;
  }

}
